/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.tablet;

import java.io.BufferedReader;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.StringReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.GlobalConstants;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.io.NWritable;
import com.nhncorp.neptune.common.io.NWritableUtils;
import com.nhncorp.neptune.fs.GPath;
import com.nhncorp.neptune.fs.NeptuneFileSystem;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.client.LockServiceClientFactory;
import com.nhncorp.pleiades.client.PleiadesUser;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.protocol.LockMetaData;


/**
 * Value Object which store Table's schema
 * @author nhn
 *
 */
public class TableSchema implements NWritable, Comparable<TableSchema> {
  public static final Log LOG = LogFactory.getLog(TableSchema.class.getName());
  public static final int DEFAULT_NUM_VERSION = 3;
  
  public static final String DIR_TABLE = "/table/";
  
	private String tableName;
  private String description;
  private List<String> columns;
	private int numOfVersion = DEFAULT_NUM_VERSION;
  private String owner;
  
  //userId -> w|r
  private Map<String, String> permissions = new HashMap<String, String>(); 
  
	public TableSchema() {
    this("", "");
	}
	
  public TableSchema(String tableName) {
    this(tableName, "");
  }

  public TableSchema(String tableName, String description) {
    this(tableName, description, new ArrayList<String>());
  }

  public TableSchema(String tableName, String description, List<String> columns) {
    this.tableName = tableName;
    this.description = description;
    this.columns = columns;
  }
	
	public TableSchema(String tableName, String description, String[] columns) {
		this.tableName = tableName;
		this.description = description;
		this.columns = new ArrayList<String>();
		
		for(String column: columns) {
			this.columns.add(column.trim());
		}
	}
	
  public void addColumn(String column) {
    columns.add(column);
  }
  
	public List<String> getColumns() {
		return columns;
	}
  
	public void setColumns(List<String> columns) {
		this.columns = columns;
	}
  
	public String getDescription() {
		return description;
	}
	public void setDescription(String description) {
		this.description = description;
	}
	public String getTableName() {
		return tableName;
	}
	public void setTableName(String tableName) {
		this.tableName = tableName;
	}
	
  public int getNumOfVersion() {
    return numOfVersion;
  }

  public void setNumOfVersion(int numOfVeriosn) {
    this.numOfVersion = numOfVeriosn;
  }

  public int hashCode() {
    return tableName.hashCode();
  }
  
  public boolean equals(Object obj) {
    if(obj == null) {
      return false;
    }
    
    if(!(obj instanceof TableSchema)) {
      return false;
    }
    
    return tableName.equals(((TableSchema)obj).tableName);
  }
  
  public static TableSchema loadTableSchema(NConfiguration conf, LockService lockService, String tableName) throws IOException {
    String schemaLocation = conf.get("table.schema.type", "filesystem");
    if(schemaLocation.equals("pleiades")) {
      return loadTableSchemaFromLockService(lockService, tableName);
    } else {
      return loadTableSchemaFromFile(conf, tableName);
    }
  }
  
  private static TableSchema loadTableSchemaFromFile(NConfiguration conf, String tableName) throws IOException {
    if(tableName.equals(Constants.TABLE_NAME_ROOT)) {
      return GlobalConstants.ROOT_TABLE;
    } else if(tableName.equals(Constants.TABLE_NAME_META)) {
      return GlobalConstants.META_TABLE;
    }
    TableSchema tableSchema = new TableSchema(tableName);
    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
    GPath schemaPath = getTableSchemaPath(conf, tableName);
    if(!fs.exists(schemaPath)) {
      return null;
    }
    GPath[] columnPaths = fs.list(schemaPath);
    
    if(columnPaths == null || columnPaths.length == 0) {
      return null;
    }
    
    //get numOfVersion, owner
    GPath schemaInfoPath = new GPath(schemaPath, Constants.SCHEMA_INFO_FILENAME);
    BufferedReader infoReader = null;
    try {
      infoReader = new BufferedReader(new InputStreamReader(fs.open(schemaInfoPath)));
      String line = infoReader.readLine();
      if(line == null) {
        return null;
      }
      tableSchema.setNumOfVersion(Integer.parseInt(line));
      
      line = infoReader.readLine();
      if(line == null) {
        return null;
      }
      
      tableSchema.setOwnerInfo(line);
    } finally {
      if(infoReader != null) {
        infoReader.close();
      }
    }
    
    //get description
    GPath descriptionPath = new GPath(schemaPath, Constants.SCHEMA_DESCRIPTION_FILE_NAME);
    BufferedReader reader = null;
    StringBuffer sb = new StringBuffer(100);
    try {
      reader = new BufferedReader(new InputStreamReader(fs.open(descriptionPath)));
      String line = null;
      while( (line = reader.readLine()) != null) {
        sb.append(line).append("\n");
      }
    } finally {
      if(reader != null) {
        reader.close();
      }
    }
    tableSchema.setDescription(sb.toString());
    
    //get columns
    for(GPath columnPath: columnPaths) {
      if(fs.isDirectory(columnPath)) {
        tableSchema.addColumn(columnPath.getName());
      }
    }

    return tableSchema;
  } 
  
  private static TableSchema loadTableSchemaFromLockService(LockService lockService, String tableName) throws IOException {
    if(lockService == null) {
      return null;
    }
    
    String tableLockPath = Constants.PATH_SCHEMA + "/" + tableName;
    
    TableSchema table = new TableSchema(tableName);
    
    LockMetaData detailMetaData = lockService.getMetaData(tableLockPath + Constants.SCHEMA_DETAIL);
    if(detailMetaData != null) {
      String key = detailMetaData.getContentKey();
      try {
        table.setNumOfVersion(Integer.parseInt(key));
      } catch (Exception e) {
        LOG.warn("Error while get num of versions from contents key:" + e.getMessage());
      }
    }
    
    String ownerInfo = lockService.getContents(tableLockPath + Constants.SCHEMA_DETAIL);
    table.setOwnerInfo(ownerInfo);
    
    String descPath = tableLockPath + "/" + Constants.SCHEMA_DESCRIPTION_FILE_NAME;
    table.setDescription(lockService.getContents(descPath));

    Map<String, LockMetaData> columnMetaDatas = null;
    try {
      columnMetaDatas = lockService.readDir(tableLockPath);
    } catch (Exception e) {
      LOG.error(e.getMessage());
      return null;
    }

    if(columnMetaDatas == null) {
      return null;
    }
    
    SortedMap<Integer, String> columns = new TreeMap<Integer, String>();
    for(LockMetaData eachColumnLockMetaData: columnMetaDatas.values()) {
      String columnLockPath = eachColumnLockMetaData.getLockId();
      if(columnLockPath.equals(tableLockPath + Constants.SCHEMA_DETAIL)) {
        continue;
      }
      String columnName = columnLockPath.substring(tableLockPath.length() + 1);
      if(!columnLockPath.equals(descPath)) { 
        Integer columnIndex = new Integer(eachColumnLockMetaData.getContentKey());
        columns.put(columnIndex, columnName);
      }
    }
    
    for(String column: columns.values()) {
      table.addColumn(column);
    }
    return table;
  } 
  
  public static GPath getTableSchemaPath(NConfiguration conf, String tableName) {
    return new GPath(conf.get("neptune.root") + Constants.PATH_SCHEMA + "/" + tableName);
  }
  
	private void setOwnerInfo(String ownerInfo) throws IOException {
    if(ownerInfo == null || ownerInfo.trim().length() == 0) {
      //LOG.fatal("No owner info:" + tableName);
      return;
    }

    //1 line: owner
    //2 ~ n line: user\tpermission(r/w)
    BufferedReader reader = new BufferedReader(new StringReader(ownerInfo));
    String line = null;
    
    int count = 0;
    while( (line = reader.readLine()) != null) {
      if(count == 0) {
        this.owner = line;
      } else {
        String[] tokens = line.split("\t");
        if(tokens.length == 2) {
          permissions.put(tokens[0], tokens[1]);
        }
      }
      count++;
    }
  }
  
  public String getOwnerInfo() throws IOException {
    StringBuffer sb = new StringBuffer();
    
    if(owner == null) {
      sb.append("\n");
    } else {
      sb.append(owner).append("\n");
    }
    
    for(Map.Entry<String, String> entry: permissions.entrySet()) {
      sb.append(entry.getKey()).append("\t").append(entry.getValue()).append("\n");
    }
    
    return sb.toString();
  }

  public void readFields(DataInput in) throws IOException {
		tableName = NWritableUtils.readString(in);
		description = NWritableUtils.readString(in);
    numOfVersion = in.readInt();
    owner = NWritableUtils.readString(in);
		int count = in.readInt();
		for(int i = 0; i < count; i++) {
			columns.add(NWritableUtils.readString(in));
		}
	}

	public void write(DataOutput out) throws IOException {
		NWritableUtils.writeString(out, tableName);
		NWritableUtils.writeString(out, description);
		out.writeInt(numOfVersion);
    NWritableUtils.writeString(out, owner);
		int size = columns.size();
		out.writeInt(size);
		for(String column: columns) {
			NWritableUtils.writeString(out, column);
		}
	}

  public String getColumn(int i) {
    String[] columnArray = columns.toArray(new String[]{});
    return columnArray[i];
  }
  
  public String[] getColumnsArray() {
    String[] columnArray = columns.toArray(new String[]{});
    return columnArray;
  }

  public boolean containsColumn(String[] targetColumns) {
    if(targetColumns == null || targetColumns.length == 0) {
      return false;
    }
    
    Set<String> columnSet = new HashSet<String>();
    columnSet.addAll(columns);
    for(String eachColumn: targetColumns) {
      if(!columnSet.contains(eachColumn)) {
        return false;
      }
    }
    return true;
  }
  
  public String toString() {
    return tableName;
  }
  
  public void print() {
    System.out.println("Table name: " + tableName);
    System.out.println("Columns:");
    for(String column: columns) {
      System.out.println("\t" + column);
    }
  }
  
//  public static void main(String[] args) throws Exception {
//    if(args.length < 1) {
//      System.out.println("Usage: TableInfo <table name>");
//      System.exit(0);
//    }
//    long startTime = System.currentTimeMillis();
//    NConfiguration conf = new NConfiguration();
//    LockService lockService = LockServiceClientFactory.newInstance(
//        new PleiadesUser(conf.get("pleiades.service.name", "neptune"), "Test"), 
//        new PleiadesConfiguration());
//    TableSchema tableSchema = loadTableSchema(conf, lockService, args[0]);
//    System.out.println("Schema loading time: " + (System.currentTimeMillis() - startTime));
//    tableSchema.print();
//  }

  public static String getTableDataPath(NConfiguration conf, String tableName) {
    return conf.get("neptune.root") + DIR_TABLE + tableName;
  }

  public static String getTableDataTrashPath(NConfiguration conf, String tableName) {
    return conf.get("neptune.root") + "/trash" + DIR_TABLE + System.currentTimeMillis() + "/" + tableName;
  }

  public int compareTo(TableSchema o) {
    return tableName.compareTo(o.tableName);
  }

  public String getTableDetail() {
    return "owner=" + owner + "; # versions=" + numOfVersion;
  }

  public String getOwner() {
    return owner;
  }

  public void setOwner(String owner) {
    this.owner = owner;
  }

  public boolean hasPermission(String userId, String readWrite) {
    //LOG.fatal("Owner:" + owner + "," + userId);
    if(userId.equals(owner)) {
      return true;
    }
    
    String permission = permissions.get("*");
    if(permission == null) {
      permission = permissions.get(userId);
    }
    
    if(permission == null) {
      return false;
    }
    
    if("w".equals(permission)) {
      return true;
    } else {
      //권한 설정은 read이고 request는 write 요청이 온 경우
      if("w".equals(readWrite)) {
        return false;
      } else {
        return true;
      }
    }
  }

  public void addPermission(String userId, String permission) {
    permissions.put(userId, permission);
  }

  public void removePermission(String userId) {
    permissions.remove(userId);
  }
}
